/*
 *  AsyncBuffer.cpp
 *  MusicAnalyzer
 *
 *  Created by jiro on 12/03/27.
 *  Copyright 2012 jiro music. All rights reserved.
 *
 */

#include "AsyncBufferSystem.h"

#include <deque>
#include <iostream>

using namespace std;

namespace sc
{
	bool AsyncBufferSystem::continuePolling = false;

	AsyncBufferSystem::AsyncBufferSystem( int bufferByteCount, int bufferCount )
		: bufferByteCount( bufferByteCount )
	{
		Alloc( bufferByteCount, bufferCount );
		pthread_mutex_init( & mutex, NULL );
	}

	AsyncBufferSystem::~AsyncBufferSystem()
	{
		for ( int i = 0, count = rawBufferHeadPtrs.size(); i < count; i++ )
		{
			delete [] rawBufferHeadPtrs[ i ];
		}
		for ( int i = 0, count = buffers.size(); i < count; i++ )
		{
			delete [] buffers[ i ];
		}
	}

	void AsyncBufferSystem::StartPolling()
	{
		continuePolling = true;
		pthread_create( & pollingThread, NULL, ReadData, this );
	}

	void AsyncBufferSystem::StopPolling()
	{
		//	wait for all data are polled
		while ( not dirtyBufferQueue.empty() )
		{
			usleep( 10 );
		}
		continuePolling = false;
		pthread_join( pollingThread, NULL );
	}

	bool AsyncBufferSystem::AddObserver( IObserver< Buffer > & observer )
	{
		if ( continuePolling )
		{
			return false;
		}
		observers.push_back( & observer );
		return true;
	}

	void AsyncBufferSystem::Alloc( int bufferByteCount, int bufferCount )
	{
		Buffer* newBuffers = new Buffer[ bufferCount ];
		uint8_t* ptr = new uint8_t[ bufferCount * bufferByteCount ];
		rawBufferHeadPtrs.push_back( ptr );
		for ( int i = 0; i < bufferCount; i++ )
		{
			newBuffers[ i ].Set( ptr, bufferByteCount );
			ptr += bufferByteCount;
			cleanBufferStack.push( & newBuffers[ i ] );
		}
		buffers.push_back( newBuffers );
	}

	int AsyncBufferSystem::GetBufferCount() const
	{
		return buffers.size();
	}

	int AsyncBufferSystem::GetBufferByteCount() const
	{
		return bufferByteCount;
	}

	void AsyncBufferSystem::WriteData( const uint8_t* data, int byteCount )
	{
		Buffer* buffer = NULL;
		pthread_mutex_lock( & mutex );
		{
			if ( cleanBufferStack.empty() )
			{
				Alloc( bufferByteCount, buffers.size() );
			}
			buffer = cleanBufferStack.top();
			cleanBufferStack.pop();
		}
		pthread_mutex_unlock( & mutex );

//		cout << "stack: " << cleanBufferStack.size() << "\n";
//		cout << buffer << ", " << (int) *data << "\n";

		buffer->WriteData( data, byteCount );

		pthread_mutex_lock( & mutex );
		{
			dirtyBufferQueue.push( buffer );
		}
		pthread_mutex_unlock( & mutex );
	}

	void* AsyncBufferSystem::ReadData( void* arg )
	{
		AsyncBufferSystem* sys = reinterpret_cast< AsyncBufferSystem* >( arg );
		const int notifyCount = sys->observers.size();
		while ( continuePolling )
		{
			Buffer* buffer = NULL;
			pthread_mutex_lock( & sys->mutex );
			{
				if ( sys->dirtyBufferQueue.empty() )
				{
					pthread_mutex_unlock( & sys->mutex );
					continue;
				}
				buffer = sys->dirtyBufferQueue.front();
				sys->dirtyBufferQueue.pop();
			}
			pthread_mutex_unlock( & sys->mutex );

			for ( int i = 0; i < notifyCount; i++ )
			{
//				cout << hex << buffer << endl;
				( ( sys->observers )[ i ] )->Notify( buffer );
			}

			pthread_mutex_lock( & sys->mutex );
			{
				sys->cleanBufferStack.push( buffer );
			}
			pthread_mutex_unlock( & sys->mutex );

		}
		pthread_exit( arg );		//	arg is never used...
		return NULL;
	}
}
